package com.mango.ch01;

import com.mango.HDFSTools.HDFS_HA_API;
import scala.*;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;

import com.mango.HDFSTools.HDFSAPI;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;

/**
 * 使用spark的方式，在内存中完成排序，适合数据量小的情况 本demo需要一个资源文本文件和一个脚本执行文件
 * com.mango.ch01.Spark_SecondarySortJOb
 *
 * @author Mango
 */
public class Spark_SecondarySortJOb {
    static Path outPath = new Path("/Spark_SecondarySortJOb/spark_output");
    static Path inputPath = new Path("/Spark_SecondarySortJOb/spark_input");

    @SuppressWarnings({"serial"})
    public static void main(String[] args) {
        // 读取输入参数并且验证
        /*
         * if (args.length < 1) { System.out.print("useage: SecondarySort<file>");
         * System.exit(0); } String inputPath = args[0];
         * System.out.println("args[0]: <file>=" + args[0]);
         */
        // 通过创建一个javaSparkContext对象（ctx）来连接到SparkMaster
        // 连接到Spark Master
//        chenckPath();
        final JavaSparkContext ctx = new JavaSparkContext();
        // 使用ctx创建JavaRdd《String》 输入数据的格式是 <name><,><time><,><value>
        JavaRDD<String> lines = ctx.textFile(inputPath.toString(), 1);
        // 用javaRDD<String>创建键值对，键-name 值（time，value）的javaPairRDD对象
        JavaPairRDD<String, Tuple2<Integer, Integer>> pairs = lines
                .mapToPair(new PairFunction<String, String, Tuple2<Integer, Integer>>() {
                    // 第一个参数是 T输入的数据 是String类型
                    // 第二个参数是 key String 类型
                    // 第三个参数是 tuple2<integer,integer>
                    @Override
                    public Tuple2<String, Tuple2<Integer, Integer>> call(String t) throws Exception {
                        // TODO Auto-generated method stub
                        String[] tokens = t.split(",");
                        String name = tokens[0];
                        for (String string : tokens) {
                            System.out.println(string + ",");

                        }
                        Integer time = new Integer(tokens[1]);
                        Integer value = new Integer(tokens[2]);
                        Tuple2<Integer, Integer> timeValue = new Tuple2<Integer, Integer>(time, value);
                        return new Tuple2<String, Tuple2<Integer, Integer>>(name, timeValue);
                    }
                });
        // 验证收集的javaPairRDD的所有值 要避免在生产集群中使用collect（）方法调试展示 影响性能
        pairs.saveAsTextFile(outPath.toString() + "/1");
        // 按键name对javaPariRDD中的元素分组 使用groupbykey 方法 执行规约数据
        JavaPairRDD<String, Iterable<Tuple2<Integer, Integer>>> groups = pairs.groupByKey();
        groups.saveAsTextFile(outPath.toString() + "/2");
        // 验证分组之后的数据
        // 对规约器中的值排序 得到最终的输出
        JavaPairRDD<String, Iterable<Tuple2<Integer, Integer>>> sorted = groups
                .mapValues(new Function<Iterable<Tuple2<Integer, Integer>>, // 输入
                        Iterable<Tuple2<Integer, Integer>>// 输出
                        >() {

                    @Override
                    public Iterable<Tuple2<Integer, Integer>> call(Iterable<Tuple2<Integer, Integer>> v1)
                            throws Exception {
                        // TODO Auto-generated method stub
                        List<Tuple2<Integer, Integer>> newList = new ArrayList<>();
                        for (Tuple2<Integer, Integer> t : v1) {
                            newList.add(t);
                        }
                        newList.sort(new myTupleComparator());// 添加到集合里边，然后使用list的排序对单个数据排序
                        return newList;
                    }
                });
        // 收集所有的JavaPairRDD的所有值 并打印出来
        sorted.saveAsTextFile(outPath.toString() + "/3");
        // 完成
        ctx.close();
        System.exit(0);
    }

    static void chenckPath() {
        // 创建输入文件夹
        Configuration conf = HDFS_HA_API.getConf();
        try {
            HDFSAPI hdfs = new HDFSAPI(conf);
            // hdfs.createDirectory(transactions);
            // 先删除已经有的输出文件夹
            hdfs.delDirectory(outPath);
            hdfs.closeFilseSystem();
            // hdfs.orpOver();
        } catch (IOException e1) {
            e1.printStackTrace();
            System.out.println("----------文件操作失败");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
    }
}
